package com.practice.gmall.realtime.joinAPI;

import com.practice.gmall.realtime.app.BaseSQLApp;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * ClassName: Tutorial_InnerJoin
 * Package: com.practice.gmall.realtime.joinAPI
 * Description: test flink inner join API
 *
 * @Author lzy.ethan@qq.com
 * @Create 2023-02-14 8:35
 */
public class Tutorial_InnerJoin extends BaseSQLApp {
    public static void main(String[] args) {
        new Tutorial_InnerJoin().init(3337,"Tutorial_InnerJoin",2);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        // 必须设置数据的ttl，在inner join中如果任意一条数据超过15秒没有使用则会把它从内存中删除
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15));

        tEnv.executeSql("create table left_tb( " +
                "    id string, " +
                "    name string " +
                ")with ( " +
                "    'connector' = 'kafka', " +
                "    'topic' = 'left_tb', " +
                "    'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092,hadoop164:9092', " +
                "    'properties.group.id' = 'Tutorial_InnerJoin', " +
                "    'scan.startup.mode' = 'latest-offset', " +
                "    'format' = 'csv' " +
                ")");

        tEnv.executeSql("create table right_tb( " +
                "    id string, " +
                "    gender string " +
                ")with ( " +
                "    'connector' = 'kafka', " +
                "    'topic' = 'right_tb', " +
                "    'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092,hadoop164:9092', " +
                "    'properties.group.id' = 'Tutorial_InnerJoin', " +
                "    'scan.startup.mode' = 'latest-offset', " +
                "    'format' = 'csv' " +
                ")");

        Table join_tb = tEnv.sqlQuery("select l.id, name, gender from left_tb l inner join right_tb  r on l.id = r.id");

        join_tb.execute().print();
    }
}
